package com.atguigu.shunt;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gmd
 * @desc 数据分流：奇数、偶数拆分成不同流
 * @since 2024-11-25 10:53:46
 */
public class OutputByFilter {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);

        DataStreamSource<String> socketDS = env.socketTextStream("127.0.0.1", 7777);

        /*
         * 使用filter来实现分流效果
         *
         * 说明：此方法通过两次调用filter函数来实现数据的分流
         * 缺点：同一个数据需要被处理两遍
         */
        SingleOutputStreamOperator<String> even = socketDS.filter(value -> Integer.parseInt(value) % 2 == 0);
        SingleOutputStreamOperator<String> odd = socketDS.filter(value -> Integer.parseInt(value) % 2 == 1);

        even.print("偶数流：");
        odd.print("奇数流：");

        env.execute();
    }

}
